#!/usr/bin/env python3
"""
验证requirements.txt文件的完整性和质量
基于requirements-dev.txt的成功验证模式
"""

import os
import sys
import re
import subprocess
import tempfile
from pathlib import Path
from typing import Dict, List, Tuple, Set, Any
from collections import defaultdict

def print_header(title: str):
    """打印格式化的标题"""
    print(f"\n{'='*60}")
    print(f" {title}")
    print(f"{'='*60}")

def print_section(title: str):
    """打印章节标题"""
    print(f"\n{'-'*40}")
    print(f" {title}")
    print(f"{'-'*40}")

def validate_syntax(requirements_file: str) -> Tuple[bool, List[str]]:
    """验证requirements.txt语法"""
    print_section("1. 语法验证")
    
    errors = []
    line_number = 0
    
    try:
        with open(requirements_file, 'r', encoding='utf-8') as f:
            for line_number, line in enumerate(f, 1):
                line = line.strip()
                
                # 跳过空行和注释
                if not line or line.startswith('#'):
                    continue
                
                # 检查基本格式
                if '==' in line:
                    # 固定版本格式检查
                    if not re.match(r'^[a-zA-Z0-9_\-\[\]]+==[\d\.\w\-\+]+$', line):
                        errors.append(f"行 {line_number}: 固定版本格式错误: {line}")
                elif '>=' in line and '<' in line:
                    # 版本范围格式检查
                    if not re.match(r'^[a-zA-Z0-9_\-\[\]]+>=[\d\.\w\-\+]+,<[\d\.\w\-\+]+$', line):
                        errors.append(f"行 {line_number}: 版本范围格式错误: {line}")
                else:
                    # 其他格式检查
                    if not re.match(r'^[a-zA-Z0-9_\-\[\]]+[><=!]+[\d\.\w\-\+,<>=\s]*$', line):
                        errors.append(f"行 {line_number}: 依赖格式错误: {line}")
        
        if errors:
            print("❌ 语法验证失败:")
            for error in errors:
                print(f"  {error}")
            return False, errors
        else:
            print("✅ 语法验证通过")
            return True, []
            
    except Exception as e:
        error_msg = f"读取文件失败: {str(e)}"
        print(f"❌ {error_msg}")
        return False, [error_msg]

def check_duplicates(requirements_file: str) -> Tuple[bool, List[str]]:
    """检查重复依赖"""
    print_section("2. 重复依赖检查")
    
    packages = defaultdict(list)
    line_number = 0
    
    try:
        with open(requirements_file, 'r', encoding='utf-8') as f:
            for line_number, line in enumerate(f, 1):
                line = line.strip()
                
                # 跳过空行和注释
                if not line or line.startswith('#'):
                    continue
                
                # 提取包名
                package_name = re.split(r'[><=!]', line)[0].strip()
                if '[' in package_name:
                    package_name = package_name.split('[')[0]
                
                packages[package_name].append(line_number)
        
        duplicates = []
        for package, lines in packages.items():
            if len(lines) > 1:
                duplicates.append(f"包 '{package}' 在行 {', '.join(map(str, lines))} 重复出现")
        
        if duplicates:
            print("❌ 发现重复依赖:")
            for duplicate in duplicates:
                print(f"  {duplicate}")
            return False, duplicates
        else:
            print("✅ 无重复依赖")
            return True, []
            
    except Exception as e:
        error_msg = f"检查重复依赖失败: {str(e)}"
        print(f"❌ {error_msg}")
        return False, [error_msg]

def check_version_strategy(requirements_file: str) -> Tuple[bool, List[str]]:
    """检查版本策略一致性"""
    print_section("3. 版本策略检查")
    
    version_patterns = {
        'fixed': 0,      # ==x.y.z
        'compatible': 0,  # >=x.y.z,<major+1.0.0
        'other': 0       # 其他格式
    }
    
    issues = []
    
    try:
        with open(requirements_file, 'r', encoding='utf-8') as f:
            for line_number, line in enumerate(f, 1):
                line = line.strip()
                
                # 跳过空行和注释
                if not line or line.startswith('#'):
                    continue
                
                if '==' in line and ',' not in line:
                    version_patterns['fixed'] += 1
                elif '>=' in line and '<' in line:
                    version_patterns['compatible'] += 1
                else:
                    version_patterns['other'] += 1
        
        total_deps = sum(version_patterns.values())
        
        print(f"版本策略统计:")
        print(f"  固定版本 (==): {version_patterns['fixed']} ({version_patterns['fixed']/total_deps*100:.1f}%)")
        print(f"  兼容版本 (>=,<): {version_patterns['compatible']} ({version_patterns['compatible']/total_deps*100:.1f}%)")
        print(f"  其他格式: {version_patterns['other']} ({version_patterns['other']/total_deps*100:.1f}%)")
        
        # 推荐使用兼容版本策略
        if version_patterns['compatible'] >= total_deps * 0.8:
            print("✅ 版本策略良好 (主要使用兼容版本范围)")
            return True, []
        elif version_patterns['fixed'] >= total_deps * 0.8:
            issues.append("建议使用兼容版本范围 (>=x.y.z,<major+1.0.0) 而非固定版本")
            print("⚠️  版本策略可优化 (过多固定版本)")
            return True, issues
        else:
            issues.append("版本策略不一致，建议统一使用兼容版本范围")
            print("❌ 版本策略混乱")
            return False, issues
            
    except Exception as e:
        error_msg = f"检查版本策略失败: {str(e)}"
        print(f"❌ {error_msg}")
        return False, [error_msg]

def check_dev_dependencies(requirements_file: str) -> Tuple[bool, List[str]]:
    """检查是否包含开发依赖"""
    print_section("4. 开发依赖分离检查")
    
    # 常见开发工具列表
    dev_tools = {
        'pytest', 'pytest-asyncio', 'pytest-cov', 'pytest-mock',
        'black', 'flake8', 'mypy', 'isort', 'autopep8',
        'pre-commit', 'bandit', 'safety',
        'sphinx', 'sphinx-rtd-theme', 'mkdocs', 'mkdocs-material',
        'jupyter', 'ipykernel', 'ipywidgets', 'notebook',
        'coverage', 'tox', 'nox'
    }
    
    found_dev_deps = []
    
    try:
        with open(requirements_file, 'r', encoding='utf-8') as f:
            for line_number, line in enumerate(f, 1):
                line = line.strip()
                
                # 跳过空行和注释
                if not line or line.startswith('#'):
                    continue
                
                # 提取包名
                package_name = re.split(r'[><=!]', line)[0].strip()
                if '[' in package_name:
                    package_name = package_name.split('[')[0]
                
                if package_name.lower() in dev_tools:
                    found_dev_deps.append(f"行 {line_number}: {package_name}")
        
        if found_dev_deps:
            print("⚠️  发现开发依赖，建议移至requirements-dev.txt:")
            for dep in found_dev_deps:
                print(f"  {dep}")
            return True, [f"发现 {len(found_dev_deps)} 个开发依赖"]
        else:
            print("✅ 无开发依赖混入")
            return True, []
            
    except Exception as e:
        error_msg = f"检查开发依赖失败: {str(e)}"
        print(f"❌ {error_msg}")
        return False, [error_msg]

def check_security(requirements_file: str) -> Tuple[bool, List[str]]:
    """安全漏洞检查"""
    print_section("5. 安全漏洞检查")
    
    try:
        # 检查是否安装了safety
        result = subprocess.run(['safety', '--version'], 
                              capture_output=True, text=True, timeout=10)
        if result.returncode != 0:
            print("⚠️  safety未安装，跳过安全检查")
            return True, ["safety工具未安装"]
        
        # 运行安全检查
        result = subprocess.run(['safety', 'check', '-r', requirements_file], 
                              capture_output=True, text=True, timeout=60)
        
        if result.returncode == 0:
            print("✅ 无已知安全漏洞")
            return True, []
        else:
            vulnerabilities = result.stdout.strip().split('\n')
            print("❌ 发现安全漏洞:")
            for vuln in vulnerabilities[-10:]:  # 显示最后10行
                if vuln.strip():
                    print(f"  {vuln}")
            return False, vulnerabilities
            
    except subprocess.TimeoutExpired:
        error_msg = "安全检查超时"
        print(f"⚠️  {error_msg}")
        return True, [error_msg]
    except Exception as e:
        error_msg = f"安全检查失败: {str(e)}"
        print(f"⚠️  {error_msg}")
        return True, [error_msg]

def check_installability(requirements_file: str) -> Tuple[bool, List[str]]:
    """检查依赖可安装性"""
    print_section("6. 依赖可安装性检查")
    
    try:
        # 创建临时虚拟环境进行测试
        with tempfile.TemporaryDirectory() as temp_dir:
            venv_path = os.path.join(temp_dir, 'test_venv')
            
            # 创建虚拟环境
            result = subprocess.run([sys.executable, '-m', 'venv', venv_path], 
                                  capture_output=True, text=True, timeout=30)
            if result.returncode != 0:
                error_msg = f"创建虚拟环境失败: {result.stderr}"
                print(f"❌ {error_msg}")
                return False, [error_msg]
            
            # 确定pip路径
            if os.name == 'nt':  # Windows
                pip_path = os.path.join(venv_path, 'Scripts', 'pip')
            else:  # Unix/Linux/macOS
                pip_path = os.path.join(venv_path, 'bin', 'pip')
            
            # 升级pip
            subprocess.run([pip_path, 'install', '--upgrade', 'pip'], 
                         capture_output=True, timeout=60)
            
            # 尝试安装依赖（仅检查，不实际安装）
            result = subprocess.run([pip_path, 'install', '--dry-run', '-r', requirements_file], 
                                  capture_output=True, text=True, timeout=120)
            
            if result.returncode == 0:
                print("✅ 所有依赖可正常安装")
                return True, []
            else:
                errors = result.stderr.strip().split('\n')
                print("❌ 依赖安装检查失败:")
                for error in errors[-5:]:  # 显示最后5行错误
                    if error.strip():
                        print(f"  {error}")
                return False, errors
                
    except subprocess.TimeoutExpired:
        error_msg = "依赖安装检查超时"
        print(f"⚠️  {error_msg}")
        return True, [error_msg]
    except Exception as e:
        error_msg = f"依赖安装检查失败: {str(e)}"
        print(f"⚠️  {error_msg}")
        return True, [error_msg]

def check_file_structure(requirements_file: str) -> Tuple[bool, List[str]]:
    """检查文件结构和组织"""
    print_section("7. 文件结构检查")
    
    issues = []
    sections = []
    current_section = None
    
    try:
        with open(requirements_file, 'r', encoding='utf-8') as f:
            for line_number, line in enumerate(f, 1):
                line = line.strip()
                
                # 检查章节标题
                if line.startswith('# ===='):
                    current_section = line
                    sections.append(current_section)
                elif line.startswith('#') and current_section:
                    # 章节内的注释
                    continue
                elif line and not line.startswith('#'):
                    # 依赖行
                    if not current_section:
                        issues.append(f"行 {line_number}: 依赖 '{line}' 未归类到任何章节")
        
        print(f"发现 {len(sections)} 个功能章节")
        
        if len(sections) >= 10:
            print("✅ 文件结构良好，按功能模块组织")
        elif len(sections) >= 5:
            print("⚠️  文件结构可接受，建议进一步细分")
        else:
            issues.append("建议按功能模块组织依赖")
            print("❌ 文件结构需要改进")
        
        if issues:
            for issue in issues:
                print(f"  {issue}")
            return len(issues) == 0, issues
        else:
            return True, []
            
    except Exception as e:
        error_msg = f"检查文件结构失败: {str(e)}"
        print(f"❌ {error_msg}")
        return False, [error_msg]

def generate_summary_report(results: Dict[str, Tuple[bool, List[str]]]) -> Dict[str, Any]:
    """生成验证总结报告"""
    print_header("验证总结报告")
    
    total_checks = len(results)
    passed_checks = sum(1 for success, _ in results.values() if success)
    
    # 统计问题
    critical_issues = []
    warnings = []
    
    for check_name, (success, issues) in results.items():
        if not success:
            critical_issues.extend(issues)
        elif issues:  # 成功但有警告
            warnings.extend(issues)
    
    # 计算质量评分
    base_score = (passed_checks / total_checks) * 100
    penalty = min(len(critical_issues) * 10 + len(warnings) * 2, 30)
    quality_score = max(base_score - penalty, 0)
    
    # 确定质量等级
    if quality_score >= 90:
        quality_grade = "优秀"
        grade_emoji = "🏆"
    elif quality_score >= 80:
        quality_grade = "良好"
        grade_emoji = "✅"
    elif quality_score >= 70:
        quality_grade = "可接受"
        grade_emoji = "⚠️"
    else:
        quality_grade = "需要改进"
        grade_emoji = "❌"
    
    print(f"\n📊 验证统计:")
    print(f"  总检查项: {total_checks}")
    print(f"  通过检查: {passed_checks}")
    print(f"  通过率: {passed_checks/total_checks*100:.1f}%")
    print(f"  严重问题: {len(critical_issues)}")
    print(f"  警告信息: {len(warnings)}")
    print(f"  质量评分: {quality_score:.1f}/100")
    print(f"  质量等级: {grade_emoji} {quality_grade}")
    
    if critical_issues:
        print(f"\n❌ 严重问题 ({len(critical_issues)}):")
        for i, issue in enumerate(critical_issues[:10], 1):
            print(f"  {i}. {issue}")
        if len(critical_issues) > 10:
            print(f"  ... 还有 {len(critical_issues) - 10} 个问题")
    
    if warnings:
        print(f"\n⚠️  警告信息 ({len(warnings)}):")
        for i, warning in enumerate(warnings[:10], 1):
            print(f"  {i}. {warning}")
        if len(warnings) > 10:
            print(f"  ... 还有 {len(warnings) - 10} 个警告")
    
    return {
        'total_checks': total_checks,
        'passed_checks': passed_checks,
        'pass_rate': passed_checks/total_checks*100,
        'critical_issues': len(critical_issues),
        'warnings': len(warnings),
        'quality_score': quality_score,
        'quality_grade': quality_grade,
        'success': len(critical_issues) == 0
    }

def main():
    """主验证流程"""
    print_header("Requirements.txt 验证工具")
    
    # 确定requirements.txt路径
    script_dir = Path(__file__).parent
    project_root = script_dir.parent.parent
    requirements_file = project_root / 'requirements.txt'
    
    if not requirements_file.exists():
        print(f"❌ 文件不存在: {requirements_file}")
        sys.exit(1)
    
    print(f"📁 验证文件: {requirements_file}")
    
    # 执行所有验证检查
    results = {}
    
    # 1. 语法验证
    results['语法验证'] = validate_syntax(str(requirements_file))
    
    # 2. 重复依赖检查
    results['重复依赖检查'] = check_duplicates(str(requirements_file))
    
    # 3. 版本策略检查
    results['版本策略检查'] = check_version_strategy(str(requirements_file))
    
    # 4. 开发依赖分离检查
    results['开发依赖分离检查'] = check_dev_dependencies(str(requirements_file))
    
    # 5. 安全漏洞检查
    results['安全漏洞检查'] = check_security(str(requirements_file))
    
    # 6. 依赖可安装性检查
    results['依赖可安装性检查'] = check_installability(str(requirements_file))
    
    # 7. 文件结构检查
    results['文件结构检查'] = check_file_structure(str(requirements_file))
    
    # 生成总结报告
    summary = generate_summary_report(results)
    
    # 返回适当的退出码
    sys.exit(0 if summary['success'] else 1)

if __name__ == '__main__':
    main()